package com.atguigu.chapter05.sink;

import com.alibaba.fastjson.JSON;
import com.atguigu.bean.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisConfigBase;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.util.ArrayList;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/6/10 10:09
 */
public class Flink01_Sink_Redis_List {
    public static void main(String[] args) throws Exception {
        ArrayList<WaterSensor> waterSensors = new ArrayList<>();
        waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
        waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
        waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
        waterSensors.add(new WaterSensor("sensor_30", 1607527993000L, 10));
        waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));
        waterSensors.add(new WaterSensor("传感器", 1607527995000L, 30));
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
    
        FlinkJedisConfigBase conf = new FlinkJedisPoolConfig.Builder()
            .setHost("hadoop162")
            .setPort(6379)
            .setMaxTotal(1000)
            .setMaxIdle(10)
            .setMinIdle(1)
            .setDatabase(0)
            .setTimeout(10000)
            .build();
        env
            .fromCollection(waterSensors)
            .addSink(new RedisSink<>(conf, new RedisMapper<WaterSensor>() {
                // string(set) set(sadd) list(lpush rpush) hash(hset) zset
                // 返回一个命令描述符
                @Override
                public RedisCommandDescription getCommandDescription() {
                    // 第二个参数只对hash和zset有效, 其他的忽略
                    return new RedisCommandDescription(RedisCommand.RPUSH, null);
                }
    
                // 返回要写的数据的key
                @Override
                public String getKeyFromData(WaterSensor data) {
                    return data.getId();
                }
                
                @Override
                public String getValueFromData(WaterSensor data) {
                    return JSON.toJSONString(data);
                }
            }));
        
        env.execute();
        
    }
}
